package com.rxjava;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;

//原始Observable遇到错误，重新订阅它期望它能正常终止.
//Retry操作符不会将原始Observable的onError通知传递给观察者，它会订阅这个Observable，再给它一次机会无错误地完成它的数据序列。
//Retry总是传递onNext通知给观察者，由于重新订阅，可能会造成数据项重复.
//这个函数返回一个布尔值，如果返回true，retry应该再次订阅和镜像原始的Observable，
//如果返回false，retry会将最新的一个onError通知传递给它的观察者。

//Rxjava还实现了RetryWhen操作符。和repeatWhen类似，但是一个是重复订阅(触发onComplete)，一个是错误后重复订阅(触发onError())。
public class TryDemo {

	private Observable<String> createObserver(Boolean createExcetion) {
        return Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) {
                for (int i = 1; i <= 6; i++) {
                    if (i < 3) {
                    	emitter.onNext("onNext:" + i);
                    } else if (createExcetion) {
                    	emitter.onError(new Exception("Exception"));
                    } else {
                    	emitter.onError(new Throwable("Throw error"));
                    }
                }
            }
        });
    }

	public void retry() {
		//retry(long):指定最多重新订阅的次数，如果次数超了，它不会尝试再次订阅，它会把最新的一个onError通知传递给它的观察者。
        createObserver(false).retry(2)
        .subscribe(new Observer<String>() {
            @Override
            public void onComplete() {
                System.out.println("retry-onCompleted\n");
            }

            @Override
            public void onError(Throwable e) {
            	System.out.println("retry-onError:" + e.getMessage());
            }

            @Override
            public void onNext(String s) {
            	System.out.println("retry-onNext:" + s);
            }

			@Override
			public void onSubscribe(Disposable disposable) {

			}
        });
    }

	public void retryWhenObserver() {
		//当错误发生时，retryWhen会接收onError的throwable作为参数，
		//并根据定义好的函数返回一个Observable，如果这个Observable发射一个数据，就会重新订阅,
		//否则，就将这个通知传递给观察者然后终止。
		//总共重试3次，重试间隔3000毫秒
		createObserver(false).retryWhen(new RetryWithDelay(3, 3000)).subscribe(x -> System.out.println(x));
	}

	public static void main(String[] args) {
		TryDemo tryDemo = new TryDemo();
		//tryDemo.retry();
		tryDemo.retryWhenObserver();
	}

}
